package com.atguigu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderDetail;
import com.atguigu.realtime.bean.OrderInfo;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.time.Duration;
import java.util.HashMap;

import static com.atguigu.realtime.common.Constant.TOPIC_DWD_ORDER_DETAIL;
import static com.atguigu.realtime.common.Constant.TOPIC_DWD_ORDER_INFO;

public class DwmOrderWideApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwmOrderWideApp().init(3003,1,"DwmOrderWideApp",
                "DwmOrderWideApp",TOPIC_DWD_ORDER_INFO, TOPIC_DWD_ORDER_DETAIL);
    }
    @Override
    protected void run(StreamExecutionEnvironment env,
                       HashMap<String, DataStreamSource<String>> topicToStream) {
//        topicToStream.get(TOPIC_DWD_ORDER_INFO).print("info");
//        topicToStream.get(TOPIC_DWD_ORDER_DETAIL).print("detail");

        //1.事实表的join
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims = factsJoin(topicToStream);

        //2.join维度数据
        SingleOutputStreamOperator<OrderWide> orderWideStreamWithDims = factDims(orderWideStreamWithoutDims);
        orderWideStreamWithDims.print();
        // 3.把宽表数据写入kafka

    }

    private SingleOutputStreamOperator<OrderWide> factDims(SingleOutputStreamOperator<OrderWide> orderWideStreamWithoutDims) {
        /*    每来一i条数据，都通过jdbc  去phoenix 中查找对应的维度数据：6张表
        * */
        // 因为要建立phoenix连接 ，所以不使用new MapFunction 而使用Rich版本的方法，含有open
       return orderWideStreamWithoutDims.map(new RichMapFunction<OrderWide, OrderWide>() {

           private Connection phoenixConn;

           @Override
           public void open(Configuration parameters) throws Exception {
               phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL);
           }

           @Override
           public void close() throws Exception {
               if (phoenixConn != null) {
                   phoenixConn.close();
               }
           }

           @Override
            public OrderWide map(OrderWide orderWide) throws Exception {
               // 执行6个sql去查找对应的维度数据
               // 1. 补充userInfo
               JSONObject userInfo = DimUtil.readDimFromPhoenix(phoenixConn, "dim_user_info", orderWide.getUser_id());
               orderWide.setUser_gender(userInfo.getString("GENDER"));
               orderWide.calcUserAge(userInfo.getString("BIRTHDAY"));

               // 2. 补充省份
               JSONObject provinceInfo = DimUtil.readDimFromPhoenix(phoenixConn, "dim_base_province", orderWide.getProvince_id());
               orderWide.setProvince_3166_2_code(provinceInfo.getString("ISO_3166_2"));
               orderWide.setProvince_area_code(provinceInfo.getString("AREA_CODE"));
               orderWide.setProvince_iso_code(provinceInfo.getString("ISO_CODE"));
               orderWide.setProvince_name(provinceInfo.getString("NAME"));

               // 3. 补充sku
               JSONObject skuInfo = DimUtil.readDimFromPhoenix(phoenixConn, "dim_sku_info", orderWide.getSku_id());
               orderWide.setSku_name(skuInfo.getString("SKU_NAME"));
               orderWide.setOrder_price(skuInfo.getBigDecimal("PRICE"));

               orderWide.setSpu_id(skuInfo.getLong("SPU_ID"));
               orderWide.setTm_id(skuInfo.getLong("TM_ID"));
               orderWide.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

               // 4. 补充spu
               JSONObject spuInfo = DimUtil.readDimFromPhoenix(phoenixConn, "dim_spu_info", orderWide.getSpu_id());
               orderWide.setSpu_name(spuInfo.getString("SPU_NAME"));
               // 5. 补充tm
               JSONObject tmInfo = DimUtil.readDimFromPhoenix(phoenixConn, "dim_base_trademark", orderWide.getTm_id());
               orderWide.setTm_name(tmInfo.getString("TM_NAME"));
               // 5. 补充c3
               JSONObject c3Info = DimUtil.readDimFromPhoenix(phoenixConn, "dim_base_category3", orderWide.getCategory3_id());
               orderWide.setCategory3_name(c3Info.getString("NAME"));


               return orderWide; // 如果表格未定义，就是6张表未maxwell监控
            }
        });

    }

    private SingleOutputStreamOperator<OrderWide> factsJoin(HashMap<String, DataStreamSource<String>> stream) {
        // interval join  1.只支持事件时间，2。必须keyby之后使用
        KeyedStream<OrderInfo, Long> orderInfoStream = stream
                .get(TOPIC_DWD_ORDER_INFO)
                .map(info -> JSON.parseObject(info, OrderInfo.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((info, ts) -> info.getCreate_ts())
                )
                .keyBy(OrderInfo::getId); // 这是订单的id

        KeyedStream<OrderDetail, Long> orderDetailStream = stream
                .get(TOPIC_DWD_ORDER_DETAIL)
                .map(detail -> JSON.parseObject(detail, OrderDetail.class))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((detail, ts) -> detail.getCreate_ts())
                )
                .keyBy(OrderDetail::getOrder_id);// 这里是orderId 是详情的id 一定不要写错了！！


       return orderInfoStream
                .intervalJoin(orderDetailStream)
                //设置上限和下限
                .between(Time.seconds(-5),Time.seconds(5))
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo left,
                                               OrderDetail right,
                                               Context ctx,
                                               Collector<OrderWide> out) throws Exception {
                        out.collect(new OrderWide(left,right));
                    }
                });


    }
}
